package com.shujia.web.controller;

import com.shujia.utils.Config;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.hbase.client.*;
import org.apache.hadoop.hbase.util.Bytes;
import org.springframework.web.bind.annotation.GetMapping;
import org.springframework.web.bind.annotation.RestController;

import java.io.IOException;
import java.util.HashMap;

@RestController
public class TourApi {

    private HTableInterface CityTable = null;
    private HTableInterface provinceTable = null;

    public TourApi() {

        //创建hbase链接
        //创建hadoop配置文件对象
        Configuration configuration = new Configuration();
        //指定zk地址
        configuration.set("hbase.zookeeper.quorum", Config.get("hbase.zookeeper.quorum"));

        //建立连接   和zk的连接
        HConnection connection = null;
        try {
            connection = HConnectionManager.createConnection(configuration);
            CityTable = connection.getTable(Config.get("city.tourist.index.hbase.table.name"));
            provinceTable = connection.getTable(Config.get("province.tourist.index.hbase.table.name"));

        } catch (IOException e) {
            e.printStackTrace();
        }
    }


    //http://localhost:8080/getCltyFlow?city=83401&day=20180503&column=age_flow&k=123456
    @GetMapping("getCltyFlow")
    public HashMap<String, Long> getFlow(String city, String day, String column, String k) {


        //查询数据库进行权限验证
        if (!k.equals("123456")) {
            return null;
        }


        System.out.println("city=" + city + "  day=" + day + "   column=" + column);

        //返回集合，  springboot  会自动将hashmao转换成json格式
        HashMap<String, Long> r = new HashMap<>();

        /*
         * 连接hbase查询数据
         *
         */

        String rowkey = city + "_" + day;

        Get get = new Get(rowkey.getBytes());

        get.addColumn("info".getBytes(), column.getBytes());
        Result result = null;
        if ("flow".equals(column)) {
            try {
                result = CityTable.get(get);

                byte[] bytes = result.getValue("info".getBytes(), column.getBytes());
                String value = Bytes.toString(bytes);

                r.put("flow", Long.parseLong(value));

            } catch (IOException e) {
                e.printStackTrace();
            }

            return r;

        }


        try {
            result = CityTable.get(get);
            byte[] bytes = result.getValue("info".getBytes(), column.getBytes());
            String value = Bytes.toString(bytes);

            String[] split = value.split("\\|");

            for (String kv : split) {
                String key = kv.split(":")[0];
                Long v = Long.parseLong(kv.split(":")[1]);
                r.put(key, v);
            }
        } catch (IOException e) {
            e.printStackTrace();
        }

        return r;
    }


    //http://localhost:8080/getProvinceFlow?province=834&day=20180503&column=age_flow&k=123456
    @GetMapping("getProvinceFlow")
    public HashMap<String, Long> getFlogetProvinceFloww(String province, String day, String column, String k) {


        //查询数据库进行权限验证
        if (!k.equals("123456")) {
            return null;
        }


        System.out.println("city=" + province + "  day=" + day + "   column=" + column);

        //返回集合，  springboot  会自动将hashmao转换成json格式
        HashMap<String, Long> r = new HashMap<>();

        /*
         * 连接hbase查询数据
         *
         */

        String rowkey = province + "_" + day;

        Get get = new Get(rowkey.getBytes());

        get.addColumn("info".getBytes(), column.getBytes());

        Result result = null;
        try {
            result = provinceTable.get(get);
            byte[] bytes = result.getValue("info".getBytes(), column.getBytes());
            String value = Bytes.toString(bytes);

            String[] split = value.split("\\|");

            for (String kv : split) {
                String key = kv.split(":")[0];
                Long v = Long.parseLong(kv.split(":")[1]);
                r.put(key, v);
            }
        } catch (IOException e) {
            e.printStackTrace();
        }

        return r;
    }
}
